package amq.demo;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import amq.demo.thds.AQThd;
import flib.util.TimeStr;
import flib.util.io.QSReader;

public class SentProducer{
	public static int					TASK_WAIT_TIME = 2000;
	public static String 				AQN = "WEB.ASW_Q1"; 
	public static String 				QueryTopic = "NTNU.SENT";
	public static String 				BrokerURI = "";
	public static HashMap<String,Task> 	TaskMap = new HashMap<String,Task>();
	
	public static class Task{
		public long			timeStamp;
		public long			lastRTS;
		public String 		tid;
		public String 		sent;
		public List<String> resp = new ArrayList<String>();
		
		public Task(String tid, String sent, long ts){this.tid = tid; this.sent = sent; this.timeStamp = ts;}
	}

	/**
	 * BD: Sentence Producer Example
	 * Refence:
	 * 	- http://activemq.apache.org/hello-world.html
	 * @param args
	 * 		Arg1=Broker URI
	 * 		The rest argument will be treated as query "sentence".
	 */
	public static void main(String[] args) throws Exception{
		if(args.length<2)
		{
			System.out.printf("\t[Info] Please give Argu1=Broker URI; ArguN=Query sentence file!\n");
			return;
		}
		BrokerURI = args[0];
		
		// 1) Open answer quene and listen to it
		AQThd aqt = new AQThd(AQN, BrokerURI);
		new Thread(aqt).start();
		
		// 2) Start sending sentences
		try {
            // Create a ConnectionFactory
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BrokerURI);

            // Create a Connection
            Connection connection = connectionFactory.createConnection();
            connection.start();

            // Create a Session
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Create the destination (Topic or Queue)
            Destination destination = session.createTopic(QueryTopic);

            // Create a MessageProducer from the Session to the Topic or Queue
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // Create a messages            
            TextMessage message = null;
            QSReader qsr = new QSReader(args[1]);
            qsr.skipCommentLine = qsr.skipEmptyLine = true;
            qsr.open();
            int tid=1;
            for(String line:qsr)
            {
            	//String line = new String(args[1].getBytes("UTF8"),"UTF8");
            	String tidStr = String.format("T%d", tid);
            	message = session.createTextMessage(line);
            	message.setStringProperty("UID", "John");
            	message.setStringProperty("TID", tidStr);
            	message.setStringProperty("ReplyTo", AQN);
            	message.setDoubleProperty("TimeStamp", new Date().getTime());
            	System.out.println("Sent message: "+ line + " : " + Thread.currentThread().getName());
            	// Tell the producer to send the message
                producer.send(message);                
                TaskMap.put(tidStr, new Task(tidStr, line, System.currentTimeMillis()));
                tid++;
            }          
            qsr.close();

            // Clean up
            session.close();
            connection.close();
        }
        catch (Exception e) {
            System.out.println("Caught: " + e);
            e.printStackTrace();
        }
		
		Thread.sleep(TASK_WAIT_TIME);
		Iterator<Entry<String,Task>> iter = TaskMap.entrySet().iterator();
		while(iter.hasNext())
		{
			Entry<String,Task> ety = iter.next();
			Task task = ety.getValue();
			if(task.resp.size()>0)
			{
				System.out.printf("\t%s (%s):\n", task.sent, TimeStr.ToString(task.lastRTS-task.timeStamp));
				for(String respStr:task.resp) System.out.printf("\t\t%s\n", respStr);
			}
			else
			{
				System.out.printf("\t\t%s : No response!\n", task.sent);
			}
		}
		
		System.out.printf("\t[Info] Press any key to exit...\n");
		System.in.read();
		aqt.isStop = true;
		System.out.printf("\t[Info] Exit SentProducer...\n");
	}
	
	public static void Callback(String tid, String msg)
	{
		Task task = TaskMap.get(tid);
		if(task!=null) {
			task.resp.add(msg);
			task.lastRTS = System.currentTimeMillis();
		}
	}
}
